package ck

import (
	"context"
	"flag"
	"gitee.com/yysf_xin/dplog/pkg/slog"
	"gitee.com/yysf_xin/go-common/tools"
	"github.com/ClickHouse/ch-go"
	"github.com/ClickHouse/ch-go/chpool"
	"github.com/ClickHouse/ch-go/proto"
	"net/url"
	"strings"
	"sync"
	"time"
)

type Option struct {
	Db    string `yaml:"db"`
	Batch uint   `yaml:"batch"`
	Table string `yaml:"table"`
}

type Collect struct {
	*chpool.Pool
	opt      *Option
	Total    uint
	log      *Log
	input    proto.Input
	last     time.Time
	batch    uint
	body     string
	errCount int
	lock     sync.Mutex
}

type Log struct {
	Time    proto.ColDateTime64
	Level   proto.ColStr
	Thread  proto.ColStr
	AppName proto.ColStr
	TraceID proto.ColFixedStr16
	SpanID  proto.ColFixedStr16
	Logger  proto.ColStr
	Method  proto.ColStr
	Message proto.ColStr
	Ip      proto.ColStr
}

func toInput(l *Log) proto.Input {
	return proto.Input{
		{Name: "Time", Data: &l.Time},
		{Name: "Level", Data: &l.Level},
		{Name: "Thread", Data: &l.Thread},
		{Name: "AppName", Data: &l.AppName},
		{Name: "TraceID", Data: &l.TraceID},
		{Name: "SpanID", Data: &l.SpanID},
		{Name: "Logger", Data: &l.Logger},
		{Name: "Method", Data: &l.Method},
		{Name: "_raw_log_", Data: &l.Message},
		{Name: "ipaddr", Data: &l.Ip},
	}
}
func NewCollect(opt *Option) (*Collect, error) {
	var op = ch.Options{
		Address:     "localhost:9000",
		Compression: ch.CompressionLZ4,
	}
	if opt.Batch < 0 {
		opt.Batch = 1000
	}
	u, err := url.Parse(opt.Db)
	if err != nil {
		return nil, err
	}
	op.Address = u.Host
	op.User = u.User.Username()
	op.Password, _ = u.User.Password()
	op.Database = u.Path
	if strings.HasPrefix(op.Database, "/") {
		op.Database = op.Database[1:]
	}
	ctx := context.Background()
	c, err := chpool.New(ctx, chpool.Options{
		ClientOptions: op,
	})
	if err != nil {
		return nil, err
	}
	l := &Log{
		Time: *new(proto.ColDateTime64).WithPrecision(proto.PrecisionMilli),
	}
	in := toInput(l)
	return &Collect{
		opt:   opt,
		Pool:  c,
		log:   l,
		input: in,
		body:  in.Into(opt.Table),
		last:  time.Now(),
	}, nil
}

func (c *Collect) Flush() {
	if c.batch == 0 {
		return
	}
	err := c.Pool.Do(context.Background(), ch.Query{
		Body:  c.body,
		Input: c.input,
	})
	c.Total += c.batch
	c.batch = 0
	c.input.Reset()
	c.errCount = 0
	if err != nil {
		tools.Log.Println("clickhouse err", err)
		c.errCount++
		c.last = time.Now()
	}
}
func (c *Collect) FireMsg(msg *slog.MessageRef) {
	now := time.Now()
	du := now.Sub(c.last)
	if c.errCount > 10 {
		if du < 1*time.Minute {
			return
		}
	}
	ref := msg.GetRef().(*slog.MessageImpl)

	c.lock.Lock()
	defer c.lock.Unlock()

	c.batch++
	c.log.Time.Append(ref.Time)
	c.log.Level.Append(ref.Level)
	c.log.Thread.Append(ref.Thread)
	c.log.AppName.Append(ref.AppName)
	c.log.Logger.Append(ref.Logger)
	c.log.Method.Append(ref.Method)
	c.log.Ip.Append(msg.Meta.Ip)
	appAll(&c.log.Message, ref.Message, ref.SubMessage)
	appTrace(c.log, ref.TraceID)

	if c.batch >= c.opt.Batch || du > 10*time.Second {
		c.last = now
		c.Flush()
	}
}
func appTrace(l *Log, trace string) {
	appendTrace(&l.TraceID, trace)
	if len(trace) > 0 {
		idx := strings.Index(trace, ",")
		if idx > 16 {
			trace = trace[idx+1:]
		}
	}
	appendTrace(&l.SpanID, trace)
}
func appendTrace(c *proto.ColFixedStr16, s string) {
	v := [16]byte{}
	if len(s) < 16 {
		for i := len(s); i < 16; i++ {
			v[i] = ' '
		}
	}
	copy(v[:], s)
	c.Append(v)
}
func appAll(c *proto.ColStr, v, v2 string) {
	start := len(c.Buf)
	c.Buf = append(c.Buf, v...)
	c.Buf = append(c.Buf, v2...)
	end := len(c.Buf)
	c.Pos = append(c.Pos, proto.Position{Start: start, End: end})
}

func (c *Collect) FireMsg2(msg *slog.MessageRef) {
	c.Total++
	appendTrace(nil, msg.GetRef().GetTrace())
}

func InitFlag(opt *Option, prefix string) {
	flag.StringVar(&opt.Db, prefix+"db", "", "clickhouse dsn")
	flag.UintVar(&opt.Batch, prefix+"batch", 2000, "clickhouse batch size")
	flag.StringVar(&opt.Table, prefix+"table", "", "clickhouse table")
}
